package com.duobaoyu.dby.supply.biz.client.boundary.message;

import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.channel.adapter.enums.biz.message.MessageStatusEnum;
import com.duobaoyu.dby.common.core.constant.CommonConstants;
import com.duobaoyu.dby.common.core.constant.RequestConstants;
import com.duobaoyu.dby.common.core.constant.StringPool;
import com.duobaoyu.dby.supply.biz.client.boundary.message.bean.pojo.MessagePayload;
import com.duobaoyu.dby.supply.biz.enums.LockKeys;
import com.duobaoyu.dby.supply.biz.enums.RedisKeys;
import com.duobaoyu.dby.supply.core.cache.service.RedissonService;
import com.duobaoyu.dby.supply.core.common.util.DeflaterUtils;
import com.duobaoyu.dby.supply.core.framework.strategy.AbstractStrategy;
import com.duobaoyu.dby.supply.core.framework.util.FrameworkUtils;
import com.duobaoyu.dby.supply.db.constant.SupplyMapperConstant;
import com.duobaoyu.dby.supply.db.entity.message.ChannelMessage;
import com.duobaoyu.dby.supply.db.entity.message.ChannelMessageOrder;
import com.duobaoyu.dby.supply.db.entity.message.ChannelMessageProduct;
import com.duobaoyu.dby.supply.db.helper.DynamicTableNameHelper;
import com.duobaoyu.dby.supply.db.repository.message.ChannelMessageRepository;
import com.duobaoyu.dby.supply.db.util.OperateAuthUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.MDC;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author liyang
 * @version 1.0.0
 * @ClassName AbstractSupplyMessageHandlerStrategy
 * @Description 消息处理分发
 * @createTime 15:56:00 2023/09/27
 */
@Slf4j
public abstract class AbstractSupplyMessageHandlerStrategy extends AbstractStrategy<MessagePayload, Collection<ChannelMessage>> {

	@Resource
	private ChannelMessageRepository channelMessageRepository;
	@Resource
	private RedissonService redissonService;

	protected Set<String> messageLabels() {
		return Collections.singleton(StringPool.EMPTY);
	}

	@Override
	public final boolean support(Object... args) {
		if (ObjectUtils.isNotEmpty(args) && CollectionUtils.isNotEmpty(this.messageLabels())) {
			return Arrays.stream(args).anyMatch(item -> this.messageLabels().contains(String.valueOf(item)));
		}
		return false;
	}

	@Override
	protected final Collection<ChannelMessage> handler(MessagePayload messagePayload, Object beforeResult) {
		Class<? extends ChannelMessage> channelMessageClass = (Class) beforeResult;
		for (MessagePayload.ChannelMessage channelMessage : messagePayload.getChannelMessages()) {
			AbstractMessageTypeHandlerStrategy typeHandlerStrategy = FrameworkUtils.strategyInstance(AbstractMessageTypeHandlerStrategy.class, channelMessage.getMessageType());
			typeHandlerStrategy.strategyHandler(channelMessage);
		}
		List<ChannelMessage> channelMessages = this.toChannelMessages(messagePayload,channelMessageClass);
		if (!messagePayload.getIsBatch()) {
			ChannelMessage channelMessage = channelMessages.stream().findFirst().orElse(null);
			if (Objects.isNull(channelMessage)) {
				return Collections.emptyList();
			}
			String tableNameSuffix = DynamicTableNameHelper.getRequestData(SupplyMapperConstant.CHANNEL_MESSAGE_TABLE_NAME);
			String lockKey = LockKeys.CHANNEL_SYNC_MESSAGE.formatKey(tableNameSuffix, channelMessage.getUniqueKey());
			return redissonService.tryLock(RedisKeys.LOCK_REDIS.formatKey(lockKey),() -> Collections.emptyList(), () -> {
				//判断幂等
				boolean exists = this.channelMessageExists(channelMessage, channelMessageClass);
				if (exists) {
					return Collections.emptyList();
				}
				//return channelMessageRepository.save(channelMessage);
				ChannelMessage message = channelMessageRepository.save(channelMessage);
				if(log.isDebugEnabled()) {
					log.debug("ChannelMessage插入数据：{}",JSON.toJSONString(message));
				}
				return StringUtils.isNotBlank(message.getId()) ? Collections.singleton(message) : null;
			});
		} else {
			Iterator<ChannelMessage> iterator = channelMessages.iterator();
			while (iterator.hasNext()) {
				//判断幂等
				boolean exists = this.channelMessageExists(iterator.next(), channelMessageClass);
				if (exists) {
					iterator.remove();
				}
			}
			if(CollectionUtils.isEmpty(channelMessages)) {
				return Collections.emptyList();
			}
			Collection<ChannelMessage> messageList = channelMessageRepository.insertAll(channelMessages);
			if(log.isDebugEnabled()) {
				log.debug("ChannelMessage插入数据：{}",JSON.toJSONString(messageList));
			}
			return (messageList.stream().map(ChannelMessage::getId).filter(StringUtils::isBlank).count() > CommonConstants.ZERO.intValue() &&
					messageList.size() == channelMessages.size()) ? null : messageList;
			//int result = channelMessageRepository.insertBatch(channelMessages);
			//return result == channelMessages.size();
		}
	}

	@Override
	protected Object finallyHandler(MessagePayload messagePayload, Collection<ChannelMessage> channelMessages) {
		DynamicTableNameHelper.removeRequestData();
		return super.finallyHandler(messagePayload, channelMessages);
	}

	protected List<ChannelMessage> toChannelMessages(MessagePayload messagePayload, Class<? extends ChannelMessage> channelMessageClass) {
		if (CollectionUtils.isEmpty(messagePayload.getChannelMessages())) {
			return Collections.emptyList();
		}
		return messagePayload.getChannelMessages().stream()
				.filter(Objects::nonNull)
				.map(message -> {
					ChannelMessage channelMessage;
					try {
						channelMessage = channelMessageClass.newInstance();
					} catch (Exception e) {
						log.warn("ChannelMessage create object error : {}", ExceptionUtils.getStackTrace(e));
						String suffix = DynamicTableNameHelper.getRequestData(SupplyMapperConstant.CHANNEL_MESSAGE_TABLE_NAME);
						channelMessage = Objects.equals(suffix,SupplyMapperConstant.CHANNEL_MESSAGE_ORDER_SUFFIX) ? ChannelMessageOrder.builder().build() : ChannelMessageProduct.builder().build();
					}
					//channelMessage.setId(message.getMessageId());
					channelMessage.setRequestId(StringUtils.isNotBlank(MDC.get(RequestConstants.REQUEST_ID)) ? MDC.get(RequestConstants.REQUEST_ID) : StringPool.EMPTY);
					channelMessage.setChannelCode(message.getChannelCode());
					channelMessage.setConfigUniqueKey(StringUtils.isNotBlank(message.getConfigUniqueKey()) ? message.getConfigUniqueKey() : StringPool.EMPTY);
					channelMessage.setOriginalMessage(message.getOriginalMessage());
					channelMessage.setUniqueKey(message.getUniqueKey());
					channelMessage.setMessageType(message.getMessageType());
					channelMessage.setMessage(Optional.ofNullable(message.getData()).map(data -> DeflaterUtils.zipString(JSON.toJSONString(data))).orElse(StringPool.EMPTY));
					channelMessage.setStatus(MessageStatusEnum.WAIT_HANDLE.getValue());
					channelMessage.setIsBatch(message.getIsBatch());
					channelMessage.setSourceClass(message.getSourceClass());
					channelMessage.setOperationalCode(Optional.ofNullable(message.getOperationalCode()).orElse(StringPool.EMPTY));
					channelMessage.setChannelOperationalCode(Optional.ofNullable(message.getChannelOperationalCode()).orElse(StringPool.EMPTY));
					channelMessage.setRetryCount(CommonConstants.ZERO);
					channelMessage.setVersion(CommonConstants.ONE);
					channelMessage.setDelFlag(CommonConstants.ZERO);
					channelMessage.setCreateBy(OperateAuthUtil.getAdministrator());
					channelMessage.setUpdateBy(OperateAuthUtil.getAdministrator());
					channelMessage.setCreateTime(LocalDateTime.now());
					channelMessage.setUpdateTime(LocalDateTime.now());
					return channelMessage;
				})
				.collect(Collectors.toList());
	}

	private boolean channelMessageExists(ChannelMessage channelMessage,Class<? extends ChannelMessage> channelMessageClass) {
		//判断幂等
		boolean exists = channelMessageRepository.existsByUniqueKey(channelMessage.getUniqueKey(), channelMessageClass);
		if (exists) {
			if(log.isDebugEnabled()) {
				log.debug("渠道消息已存在，消息唯一标识：{}", channelMessage.getUniqueKey());
				log.debug("渠道消息已存在，消息体：{}", JSON.toJSONString(channelMessage));
			}
			//存在消息
			/*String msg = "\n" + "> - 渠道编号: " + channelMessage.getChannelCode() + "\n" +
					"> - 消息类型：" + BizMessageTypeEnum.bizMessageTypeEnum(channelMessage.getMessageType()).getDescribe() + "\n" +
					"> - 消息体：" + JSON.toJSONString(channelMessage) + "\n" +
					"> - requestId: " + MDC.get(RequestConstants.REQUEST_ID);
			DingDingUtils.sendMarkdownMsg(msg, "渠道消息幂等处理通知");*/
		}
		return exists;
	}
}
